package com.its.common.net.core.handler;

import com.its.common.net.core.INetHandlerExecutor;
import com.its.common.net.core.dto.GenericNetDTO;
import com.its.common.net.core.util.ChannelUtil;
import com.its.common.net.server.channel.ChannelHelper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.*;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.net.InetSocketAddress;

/**
 * @Author xiaxp
 * @Date 9/14/22 15:02
 * @Description
 */
@Slf4j
@ChannelHandler.Sharable
public abstract class BaseGenericHandler<T> extends SimpleChannelInboundHandler {

    protected INetHandlerExecutor netHandlerExecutor;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (acceptInboundMessage(msg)) {
            channelRead0(ctx, msg);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object obj) throws Exception {
        if (null == obj) {
            return;
        }

        T data = (T)obj;

        GenericNetDTO netDTO = GenericNetDTO.builder()
            .sender(GenericNetDTO.Sender.builder().address("").port(0).build())
            .data(data)
            .build();
        extractSendSocketInfo(ctx, netDTO);

        handleBeforeRead(ctx, obj, netDTO);

        EventExecutor channelExecutor = ctx.executor();
        Promise promise = new DefaultPromise(channelExecutor);
        Future future = netHandlerExecutor.asyncExecute(promise, this, ctx, netDTO);

        future.addListener((GenericFutureListener)f -> {
            if (f.isSuccess()) {
                if (null != netDTO.getData() && ReferenceCountUtil.refCnt(netDTO.getData()) > 1) {
                    ReferenceCountUtil.release(netDTO.getData(), 1);
                }
                handleAfterRead(ctx, netDTO);

                ctx.fireChannelRead(obj);
            }
        });
    }

    /**
     * 从参数中提取取发送端的地址和端口信息
     *
     * @param ctx 通道处理器上下文
     * @param obj 原始数据对象
     */
    protected abstract void extractSendSocketInfo(ChannelHandlerContext ctx, GenericNetDTO<T> obj);

    /**
     * 处理数据前的动作
     *
     * @param ctx  通道处理器上下文
     * @param obj  原始对象
     * @param data 数据对象
     */
    protected void handleBeforeRead(ChannelHandlerContext ctx, Object obj, GenericNetDTO<T> data) {
    }

    /**
     * 处理数据成功后的动作
     *
     * @param ctx  通道处理器上下文
     * @param data 数据对象
     */
    protected void handleAfterRead(ChannelHandlerContext ctx, GenericNetDTO<T> data) {
    }

    /**
     * 自定义数据的处理过程，也可以异步处理
     *
     * @param ctx  通道处理器上下文
     * @param data 数据对象
     */
    public abstract void handleData(ChannelHandlerContext ctx, GenericNetDTO<T> data);


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        final String addrRemote = ChannelUtil.extractRemoteAddr(ctx.channel());
        log.warn("NETTY PIPELINE: exceptionCaught exception.", cause);
        ctx.close().addListener((ChannelFutureListener)future -> {
            log.info("关闭Channel: close the connection to remote address[{}] result: {}",
                addrRemote,
                future.isSuccess());
            ChannelHelper.closeChannel(ctx.channel());
        });
    }

    protected abstract byte[] extract2bytes(T obj);

    protected byte[] byteBuf2bytes(ByteBuf buf) {
        int readableBytes = buf.readableBytes();
        byte[] bytes = new byte[readableBytes];
        buf.readBytes(bytes);
        return bytes;
    }
}
